package com.hownoon.parquet1;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.parquet.example.data.Group;

import java.io.IOException;

public class ParquetMapper  extends Mapper<Void,Group, NullWritable, LongWritable> {
    @Override
    protected void map(Void key, Group value, Context context) throws IOException, InterruptedException {
        //super.map(key, value, context);
        String bs_flag=value.getString("bs_flag", 0);
        String date=value.getString("date", 0);
        date=date.replace(" ","&");
        String price=String.valueOf(value.getFloat("price", 0));
        String volume=String.valueOf(value.getInteger("volume", 0));

        /*long pid = 0;

        try {
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(new SimpleDateFormat("yyyy-MM-dd&HH:mm:ss.SSS").parse(date));
            pid=(calendar.getTimeInMillis()/10)%100;
        } catch (ParseException e) {
            e.printStackTrace();
        }

        String part;
        if(pid<10){
            part="0"+pid;
        }else{
            part=Long.toString(pid);
        }
        String uuid = UUID.randomUUID().toString().replaceAll("-","");
        String rowKey=part+"_"+date+"_"+uuid;



        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("price"), Bytes.toBytes(price));
        put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("volume"), Bytes.toBytes(volume));
        put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("date"), Bytes.toBytes(date));
        //put.addColumn(Bytes.toBytes("Buy"), Bytes.toBytes("bs_flag"), Bytes.toBytes(bs_flag));*/
        context.write(NullWritable.get(),new LongWritable(1));

    }
}
